package com.easydev.mq.rabbit.listener;

import java.io.IOException;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.easydev.mq.exception.MessageException;
import com.easydev.mq.rabbit.RabbitMQConfig;
import com.easydev.mq.rabbit.pool.RabbitChannelPoolTool;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * <Description> rabbitMq 客户端监听类，用于监听消息，消费QUEUE中message<br>
 * 
 * @author lx<br>
 * @version 1.0<br>
 * @CreateDate 2016年3月16日 <br>
 */
abstract class RabbitMessageListener {
    private final static Logger LOGGER = LoggerFactory.getLogger(RabbitMessageListener.class);

    protected RabbitChannelPoolTool channelPool;

    // autoACK为true时，ACK在TCP层面，接受到是就算应答
    private boolean autoACK = false;

    // 设置分发方式 1、公平分发;0、轮询 默认公平
    private String distribution = "1";

    // 手动ACK时异常处理方式
    private String exception = "1";

    public RabbitMessageListener() throws Exception {
        init();
    }

    /**
     * Description: 初始化监听器<br>
     * 
     * @author lx<br>
     * @throws MessageException
     */
    private void init() throws MessageException {
        this.channelPool = RabbitChannelPoolTool.getInstance();
        if ("1".equals(RabbitMQConfig.AUTOACK)) {
            // TODO 待确认 Fair dispatch（公平分发） 必须关闭自动应答 并设置basicQos每次分发一个
            this.autoACK = true;
        }
        this.distribution = RabbitMQConfig.DISTRIBUTION;
        this.exception = RabbitMQConfig.EXCEPTION;
    }

    protected void dealMessage(String queueName, final Channel channel) throws Exception {
        // 如果queueName为空则异常
        if (StringUtils.isBlank(queueName)) {
            throw new MessageException("receive message error:::Invalid queueName,queueName can not be null");
        }
        // 设置分发方式 1、公平分发，默认轮询
        if (!"0".equals(distribution)) {
            channel.basicQos(1);
        }
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                    byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                try {
                    handleMessage(message);
                } catch (Exception e) {
                    if (!autoACK) {
                        if ("2".equals(exception)) {
                            handleFailedMessage(message);
                        } else if ("3".equals(exception)) {
                            throw new IOException("get message error");
                        } else {
                            channel.basicNack(envelope.getDeliveryTag(), false, false);
                        }
                        LOGGER.error("receive message error", e);
                        handleFailedMessage(message);
                    }
                } finally {
                    if (!autoACK) {
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            }
        };
        channel.basicConsume(queueName, autoACK, consumer);
    }

    /**
     * Description: 用于消息处理<br>
     * 
     * @author lx<br>
     * @param message <br>
     */
    public abstract void handleMessage(String message) throws Exception;

    /**
     * Description: 用于失败的消息处理<br>
     * 
     * @author lx<br>
     * @param message <br>
     */
    public abstract void handleFailedMessage(String message);
}
